import 'dart:async';

extension WithPreviousStreamExtension<T> on Stream<T> {
  Stream<(T, T?)> get withPrevious => _WithPreviousStream(this);

  Stream<(T, T)> withPreviousSeeded(T seed) => _WithPreviousStream(this, seed).cast();
}

class _WithPreviousStream<T> extends Stream<(T, T?)> {
  _WithPreviousStream(Stream<T> source, [T? seed]) {
    _previousValue = seed;

    _subscription = source.listen(
      (value) {
        _controller.add((value, _previousValue));
        _previousValue = value;
      },
      onDone: () {
        _subscription.cancel();
        _controller.close();
      },
      onError: _controller.addError,
    );
  }

  final _controller = StreamController<(T, T?)>.broadcast();

  late final StreamSubscription<T> _subscription;

  T? _previousValue;

  @override
  StreamSubscription<(T, T?)> listen(
    void Function((T, T?) event)? onData, {
    Function? onError,
    void Function()? onDone,
    bool? cancelOnError,
  }) {
    return _controller.stream.listen(
      onData,
      onError: onError,
      onDone: onDone,
      cancelOnError: cancelOnError,
    );
  }
}
